#include <string.h>
#include <stdio.h>
#include <debug_printf.h>
#include "ThmarMqttClient.h"
#include <uthash.h>
#include "sds.h"
#include "thmar_types.h"
#include "rwlock.h"

////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////// 通道订阅管理 ///////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////
// 定义ChannelContext结构体, 用于保存订阅的channel和回调函数
typedef struct ChannelContext
{
    char *channel;
    void *adapterContext;
    MessageAdapter adapter;
    UT_hash_handle hh;
    ~ChannelContext()
    {
        free(channel);
    }
} ChannelContext;

ChannelContext *channelContextHash = nullptr;
/** channelContextHash 读写锁 */
rwlock_t channelContextHashLock = RWLOCK_INITIALIZER;

/** 初始化channelContextHash */
static void initChannnelContextHash()
{
    if(channelContextHashLock.initialized == 0) {
        rwlock_init(&channelContextHashLock);
    }
    channelContextHash = nullptr;
}
/** 销毁channelContextHash */
static void destroyChannnelContextHash()
{
    rwlock_lock_write(&channelContextHashLock);
    if(channelContextHash){
        ChannelContext *channelContext,*tmp;
        HASH_ITER(hh, channelContextHash, channelContext, tmp)
        {
            HASH_DEL(channelContextHash, channelContext);
            delete channelContext;
        }
        HASH_CLEAR(hh, channelContextHash);
        channelContextHash = nullptr;
    }
    rwlock_unlock_write(&channelContextHashLock);
    if(channelContextHashLock.destroyed == 0) {
        rwlock_destroy(&channelContextHashLock);
    }
}

static ChannelContext *findChannelContext(const char *channel)
{
    rwlock_lock_read(&channelContextHashLock);
    ChannelContext *channelContext = nullptr;
    HASH_FIND_STR(channelContextHash, channel, channelContext);
    rwlock_unlock_read(&channelContextHashLock);
    return channelContext;
}

static void removeChannelContext(ChannelContext *channelContext)
{
    rwlock_lock_write(&channelContextHashLock);
    HASH_DEL(channelContextHash, channelContext);
    delete channelContext;
    rwlock_unlock_write(&channelContextHashLock);
}

static void addChannelContext(ChannelContext *channelContext)
{
    rwlock_lock_write(&channelContextHashLock);
    HASH_ADD_STR(channelContextHash, channel, channelContext);
    rwlock_unlock_write(&channelContextHashLock);
}

////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////

// 回调函数
static void onMessage(void *context, const char *topic, size_t topicLength, const char *message, size_t messageLength)
{
    COM_PRINTF_DEBUG("receive topic:%s(%d)\n", topic,topicLength);
    const char * _topic;
    /** 如果 topicLength 为 0, 则 topic 为 '\0'结尾的字符串, 直接使用 */
    /** 如果 topicLength 不为 0, 则 topic 非 '\0'结尾的字符串, 需要拷贝到新的内存中 */
    if(topicLength)
    {
        _topic = sdsnewlen(topic, topicLength);
    }else{
        _topic = topic;
    }
    COM_PRINTF_DEBUG("receive message:%s\n", message);
    /* 根据topic并查找订阅的频道, 调用消息处理函数，如果没有找到订阅的频道, 则忽略消息 */ 
    ChannelContext *channelContext = findChannelContext(_topic);
    /** 如果 topicLength 不为 0，则需要释放新分配的内存 */
    if(topicLength){
        sdsfree((sds)_topic);
    }
    if (channelContext && channelContext->adapter)
    {
        channelContext->adapter(channelContext->adapterContext, message, messageLength);
    }
}
smq::ThmarMqttClient::ThmarMqttClient()
    : dispatcherContext(nullptr), channelDispatcher(nullptr),channelLock(PTHREAD_MUTEX_INITIALIZER)
{
    pthread_mutex_init(&channelLock, nullptr);
}

smq::ThmarMqttClient::~ThmarMqttClient()
{
    pthread_mutex_destroy(&channelLock);
}
int smq::ThmarMqttClient::registerChannel(const char *channel, void *context, MessageAdapter adapter)
{
    if (channel == nullptr || adapter == nullptr)
    {
        return THMR_NULL_POINTER;
    }
    int code = THMR_OK;
    pthread_mutex_lock(&channelLock);
    ChannelContext *channelContext = findChannelContext(channel);
    if (channelContext)
    {
        channelContext->adapter = adapter;
    }
    else
    {
        COM_PRINTF_DEBUG("register channel:%s\n", channel);
        channelContext = new ChannelContext();
        channelContext->channel = strdup(channel);
        channelContext->adapterContext = context;
        channelContext->adapter = adapter;
        addChannelContext(channelContext);
        code = subscribe(channel);
        if (code != THMR_OK)
        {
            /* 订阅失败, 从哈希表中删除频道数据 */
            removeChannelContext(channelContext);
        }
    }
    pthread_mutex_unlock(&channelLock);
    return code;
}

int smq::ThmarMqttClient::unregisterChannel(const char *channel)
{
    int code = THMR_OK;
    if (!channel)
    {
        pthread_mutex_lock(&channelLock);
        ChannelContext *channelContext = findChannelContext(channel);
        if (channelContext)
        {
            code = unsubscribe(channel);
            if (code == THMR_OK)
            {
                /** 频道已经注销, 从哈希表中删除频道数据 */
                removeChannelContext(channelContext);
            }
        }
        pthread_mutex_unlock(&channelLock);
    }
    return code;
}

void smq::ThmarMqttClient::foreachChannel(void *context, NextRegisteredChannel next)
{
    if(next && channelContextHash){
        rwlock_lock_read(&channelContextHashLock);
        ChannelContext *currentContext = nullptr;
        ChannelContext *tmpContext = nullptr;
        HASH_ITER(hh, channelContextHash, currentContext, tmpContext)
        {
            next(context,currentContext->channel);
        }
        rwlock_unlock_read(&channelContextHashLock);
    }
}

int smq::ThmarMqttClient::start()
{
    initChannnelContextHash();
    int code = connect();
    if (code != THMR_OK)
    {
        return code;
    }
    code = dispatcher(this,onMessage);
    return code;
}

static void unregister_channel_callback(void *context, const char *channel)
{
    smq::ThmarMqttClient *client = (smq::ThmarMqttClient *)context;
    COM_PRINTF_DEBUG("unregister channel:%s\n", channel);
    client->unregisterChannel(channel);
}
void smq::ThmarMqttClient::stop()
{
    foreachChannel(this, unregister_channel_callback);
    destroyChannnelContextHash();
    disconnect();
}

int smq::ThmarMqttClient::dispatcher(void *context, ChannelDispatcher channelDispatcher)
{
    this->dispatcherContext = context; // 保存上下文
    this->channelDispatcher = channelDispatcher; // 保存消息处理函数
    return THMR_OK;
}

THMAR_API int thmar_mqtt_client_dispatcher(thmar_mqtt_client_t client, void *context, ChannelDispatcher dispatcher)
{
    if(!client)
    {
        return THMR_NULL_POINTER;
    }
    return ((smq::ThmarMqttClient *)client)->dispatcher(context, dispatcher);
}

THMAR_API int thmar_mqtt_client_register_channel(thmar_mqtt_client_t client, const char *channel, void *context, MessageAdapter dispatcher)
{
    
    if(!client)
    {
        return THMR_NULL_POINTER;
    }
    return ((smq::ThmarMqttClient *)client)->registerChannel(channel, context, dispatcher);
}

THMAR_API int thmar_mqtt_client_unregister_channel(thmar_mqtt_client_t client, const char *channel)
{
    if(!client)
    {
        return THMR_NULL_POINTER;
    }
    return ((smq::ThmarMqttClient *)client)->unregisterChannel(channel);
}

THMAR_API void thmar_mqtt_client_foreach_channel(thmar_mqtt_client_t client, void *context, NextRegisteredChannel next)
{
    if(!client)
    {
        return;
    }
    ((smq::ThmarMqttClient *)client)->foreachChannel(context, next);
}

THMAR_API int thmar_mqtt_client_start(thmar_mqtt_client_t client)
{
    if(!client)
    {
        return THMR_NULL_POINTER;
    }
    return ((smq::ThmarMqttClient *)client)->start();
}

THMAR_API void thmar_mqtt_client_stop(thmar_mqtt_client_t client)
{
    if(!client)
    {
        return;
    }
    ((smq::ThmarMqttClient *)client)->stop();
}
